package cn.hty.mymsg.handler.receiver.kafka;

import cn.hty.mymsg.common.domain.TaskInfo;
import cn.hty.mymsg.constans.MessageQueuePipeline;
import cn.hty.mymsg.handler.receiver.MessageReceiver;

import cn.hty.mymsg.handler.receiver.service.ConsumeService;
import cn.hty.mymsg.handler.uitls.GroupIdMappingUtils;
import cn.hutool.core.collection.CollUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

import org.jetbrains.annotations.NotNull;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.Optional;

/**
 *
 * 消费MQ的消息
 */
@Slf4j
@Component
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
@ConditionalOnProperty(name = "austin.mq.pipeline", havingValue = MessageQueuePipeline.KAFKA)
@Configuration
public class Receiver implements MessageReceiver {
    @Autowired
    private ConsumeService consumeService;
 /*   int a = 0;
    @PostConstruct
    public void A(){
        System.out.println(a++);
    }*/
    /**
     * 发送消息
     * @param consumerRecord
     * @param topicGroupId
     **/
    @KafkaListener(topics = "austinBusiness")
    public void consumer(@NotNull ConsumerRecord<?,String>consumerRecord, @Header(KafkaHeaders.GROUP_ID) String topicGroupId){
        //将信息封装为optional
        Optional<String> message = Optional.ofNullable(consumerRecord.value());
        if (message.isPresent()){
            List<TaskInfo> taskInfos =JSON.parseArray(message.get(), TaskInfo.class);
            //获取groupId，与topicGroupId相同就消费
            String messageGroupId = GroupIdMappingUtils.getGroupIdByTaskInfo(CollUtil.getFirst(taskInfos.iterator()));
            if (topicGroupId.equals(messageGroupId)){
                consumeService.consume2Send(taskInfos);
            }
        }
    }

}
